package com.atguigu.day04;

import com.atguigu.utils.IntSource;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.util.ArrayList;
import java.util.Comparator;

public class Example4 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        env
                .addSource(new IntSource())
                .keyBy(new KeySelector<Integer, String>() {
                    @Override
                    public String getKey(Integer in) throws Exception {
                        if (in % 2 == 0) {
                            return "even";
                        } else {
                            return "odd";
                        }
                    }
                })
                .process(new KeyedProcessFunction<String, Integer, String>() {
                    // {
                    //   "odd": 奇数的历史数据ListState,
                    //   "even": 偶数的历史数据ListState
                    // }
                    private ListState<Integer> history;
                    @Override
                    public void open(Configuration parameters) throws Exception {
                        history = getRuntimeContext().getListState(
                                new ListStateDescriptor<Integer>(
                                        "history",
                                        Types.INT
                                )
                        );
                    }

                    @Override
                    public void processElement(Integer in, Context ctx, Collector<String> out) throws Exception {
                        // 将in添加到in的key对应的列表状态变量中
                        history.add(in);

                        ArrayList<Integer> integers = new ArrayList<>();
                        for (Integer i : history.get()) integers.add(i);

                        // 排序
                        integers.sort(new Comparator<Integer>() {
                            @Override
                            public int compare(Integer i1, Integer i2) {
                                return i1 - i2;
                            }
                        });

                        // 格式化输出
                        StringBuilder result = new StringBuilder();
                        result.append(ctx.getCurrentKey() + " : ");
                        for (Integer i : integers) {
                            result.append(i + " => ");
                        }

                        out.collect(result.toString());
                    }
                })
                .print();

        env.execute();
    }
}
